package com.knight.hadoop.day10.mutual.fans;

import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.knight.hadoop.day09.comm.friend.QueryCommonFriendSecond;

/**
 * 
 * @author 
 * 	计算互粉
 *	输入数据：
 *  A:B,C,D,F,E,O
	B:A,C,E,K
	C:F,A,D,I
	D:A,E,F,L
	E:B,C,D,M,L
	F:A,B,C,D,E,O,M
	G:A,C,D,E,F
	H:A,C,D,E,O
	I:A,O
	J:B,O
	K:A,C,D
	L:D,E,F
	M:E,F,G
	O:A,H,I,J
 *
 */
public class MutualfansDemo {
	static class MutualfansDemoMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
		
		Text outputKey = new Text();
		IntWritable outputValue = new IntWritable(1);
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			//输入数据A:B,C,D,F,E,O
			String line = value.toString();
			String[] person_friends = line.split(":");
			String person = person_friends[0];
			String friends = person_friends[1];
			String[] friendArray = friends.split(",");
			
			String[] personAndFriend = new String[2];
			
			for(int i=0;i<friendArray.length;i++){
				personAndFriend[0] = person;
				personAndFriend[1] = friendArray[i];
				Arrays.sort(personAndFriend);
				outputKey.set(personAndFriend[0]+personAndFriend[1]);
				context.write(outputKey, outputValue);
			}
		}
	}
	
	static class MutualfansDemoReducer extends Reducer<Text, IntWritable, Text, NullWritable>{
		@Override
		protected void reduce(Text text, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
			
			int sum = 0;
			for(IntWritable intWritable:values){
				sum+=intWritable.get();
			}
			
			if(sum>=2){
				context.write(text, NullWritable.get());
			}
		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();

		Job job = Job.getInstance(configuration);

		job.setJarByClass(QueryCommonFriendSecond.class);
		
		// 设置mapper任务执行类
		job.setMapperClass(MutualfansDemoMapper.class);
		// 设置reducer任务执行类
		job.setReducerClass(MutualfansDemoReducer.class);

		// 指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		// 指定最终输出数据的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		// 指定输入参数的目录
		FileInputFormat.setInputPaths(job, new Path("E:\\home\\commonFriend\\input"));
		// 指定输出参数的目录
		FileOutputFormat.setOutputPath(job, new Path("E:\\home\\commonFriend\\MutualfansOut"));

		// 将job配置的参数以及job所用的java类所在的jar包，提交到yarn去运行
		/* job.submit(); */
		// 但是我们一般用这个，因为可以等待运行结果返回，查看运行流程
		boolean res = job.waitForCompletion(true);
		System.exit(res ? 0 : 1);
	}
}










